package drds.server.handler;

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;

import drds.server.Server;
import drds.server.config.Config;
import drds.server.config.ErrorCode;
import drds.server.datasource.GroupDataSourceNode;
import drds.server.datasource.IConnectionHolder;
import drds.server.net.packet.BinaryRowDataPacket;
import drds.server.net.packet.ErrorPacket;
import drds.server.net.packet.FieldPacket;
import drds.server.net.packet.OkPacket;
import drds.server.net.packet.RowDataPacket;
import drds.server.route.Route;
import drds.server.route.RouteNode;
import drds.server.server.Session;
import drds.server.server.SessionContext;
import drds.server.util.StringUtil;

public class SingleNodeHandler extends HandlerAdapter implements Terminatable {

	private final RouteNode node;
	private final Route route;
	private final Session session;

	// only one thread access at one time no need lock
	private volatile byte packetId;
	private volatile ByteBuffer byteBuffer;
	private volatile boolean isRunning;
	private Runnable terminateCallBack;
	private boolean prepared;
	private int fieldCount;
	private List<FieldPacket> fieldPacketList = new ArrayList<FieldPacket>();

	private volatile boolean isDefaultNodeShowTable;
	private volatile boolean isDefaultNodeShowFullTable;
	private Set<String> shardingTablesSet;

	public SingleNodeHandler(Route route, Session session) {
		if (route == null) {
			throw new NullPointerException("route");
		}
		if (session == null) {
			throw new NullPointerException("session");
		}
		this.route = route;
		this.node = route.getRouteNodes()[0];
		if (node == null) {
			throw new NullPointerException("routeNode is null!");
		}
		this.session = session;

	}

	@Override
	public void terminate(Runnable callback) {
		boolean zeroReached = false;

		if (isRunning) {
			terminateCallBack = callback;
		} else {
			zeroReached = true;
		}

		if (zeroReached) {
			callback.run();
		}
	}

	private void endRunning() {
		Runnable callback = null;
		if (isRunning) {
			isRunning = false;
			callback = terminateCallBack;
			terminateCallBack = null;
		}

		if (callback != null) {
			callback.run();
		}
	}

	private void recycleByteBuffer() {

		ByteBuffer byteBuffer = this.byteBuffer;
		if (byteBuffer != null) {
			this.byteBuffer = null;
			session.getSessionContext().recycle(this.byteBuffer);

		}
	}

	public void execute() throws Exception {
		System.currentTimeMillis();
		SessionContext sessionContext = this.session.getSessionContext();
		this.isRunning = true;
		this.packetId = 0;
		final IConnectionHolder connectionHolder = this.session.getConnectionHolder(node);
		this.node.setMasterTrue$SlaveFalse(this.route.getMasterTrue$SlaveFalse()); // 实现
		if (this.session.tryExistsCon(connectionHolder, this.node)) {
			execute(connectionHolder);
		} else {
			Config config = Server.getInstance().getConfig();
			this.node.setMasterTrue$SlaveFalse(this.route.getMasterTrue$SlaveFalse()); // 实现
			// master/slave注解
			GroupDataSourceNode groupDataSourceNode = config.getGroupDataSourceNodeMap().get(node.getName());
			groupDataSourceNode.getConnection(groupDataSourceNode.getSchemaName(), sessionContext.isAutoCommit(), node, this, node);
		}

	}

	@Override
	public void connectionAcquired(final IConnectionHolder connectionHolder) {
		this.session.bind(this.node, connectionHolder);
		execute(connectionHolder);

	}

	private void execute(IConnectionHolder connectionHolder) {
		if (this.session.closed()) {
			endRunning();
			this.session.clearResources(true);
			return;
		}
		connectionHolder.setHandler(this);
		try {
			connectionHolder.execute(this.node, this.session.getSessionContext(), session.getSessionContext().isAutoCommit());
		} catch (Exception e1) {
			executeException(connectionHolder, e1);
			return;
		}
	}

	private void executeException(IConnectionHolder connectionHolder, Exception e) {
		ErrorPacket errorPacket = new ErrorPacket();
		errorPacket.packetId = ++packetId;
		errorPacket.errorNo = ErrorCode.ERR_FOUND_EXCEPION;
		errorPacket.errorMessage = StringUtil.getBytes(e.toString());

		this.backConnectionErr(errorPacket, connectionHolder);
	}

	@Override
	public void connectionError(IConnectionHolder connectionHolder, Throwable e) {

		endRunning();
		ErrorPacket errorPacket = new ErrorPacket();
		errorPacket.packetId = ++packetId;
		errorPacket.errorNo = ErrorCode.ER_NEW_ABORTING_CONNECTION;
		errorPacket.errorMessage = StringUtil.getBytes(e.getMessage());

		SessionContext sessionContext = session.getSessionContext();
		sessionContext.write(errorPacket.write(allocByteBuffer(), sessionContext, true));
	}

	@Override
	public void errorResponse(IConnectionHolder connectionHolder, byte[] errorBytes) {
		ErrorPacket errorPacket = new ErrorPacket();
		errorPacket.read(errorBytes);
		errorPacket.packetId = ++packetId;
		backConnectionErr(errorPacket, connectionHolder);
	}

	private void backConnectionErr(ErrorPacket errorPacket, IConnectionHolder connectionHolder) {
		endRunning();

		SessionContext sessionContext = this.session.getSessionContext();
		String errorMessage = " errno:" + errorPacket.errorNo + " " + new String(errorPacket.errorMessage);
		this.session.releaseConnectionIfSafe(connectionHolder, false);

		sessionContext.setTransactionInterruptStatus(errorMessage);
		errorPacket.write(sessionContext);
		recycleByteBuffer();
	}

	/**
	 * insert/update/delete
	 * 
	 * okResponse()：读取data字节数组，组成一个OKPacket，并调用ok.write(source)
	 * 将结果写入前端连接FrontendConnection的写缓冲队列writeQueue中，
	 * 真正发送给应用是由对应的NIOSocketWR从写队列中读取ByteBuffer并返回的
	 */
	@Override
	public void okResponse(IConnectionHolder connectionHolder, byte[] okBytes) {
		boolean executeResponse = connectionHolder.syncAndExcute();
		if (executeResponse) {
			SessionContext sessionContext = session.getSessionContext();
			OkPacket okPacket = new OkPacket();
			okPacket.read(okBytes);
			boolean isCanClose2Client = false;
			if (isCanClose2Client) {
				okPacket.packetId = ++packetId;// OK_PACKET
			}

			if (isCanClose2Client) {
				session.releaseConnectionIfSafe(connectionHolder, false);
				endRunning();
			}
			okPacket.serverStatus = sessionContext.isAutoCommit() ? 2 : 1;
			recycleByteBuffer();

			if (isCanClose2Client) {
				sessionContext.setLastInsertId(okPacket.insertId);
				okPacket.write(sessionContext);
			}

		}
	}

	/**
	 * select
	 * 
	 * 行结束标志返回时触发，将EOF标志写入缓冲区，最后调用source.write(buffer)将缓冲区放入前端连接的写缓冲队列中，
	 * 等待NIOSocketWR将其发送给应用
	 */
	@Override
	public void rowEofResponse(IConnectionHolder connectionHolder, byte[] eofBytes) {

		SessionContext sessionContext = session.getSessionContext();
		session.releaseConnectionIfSafe(connectionHolder, false);
		endRunning();

		eofBytes[3] = ++packetId;
		byteBuffer = sessionContext.writeToByteBuffer(eofBytes, allocByteBuffer());
		int resultSize = sessionContext.getWriteByteBufferQueue().size() * Server.getInstance().getConfig().getDRDS().getBufferPageSize();
		resultSize = resultSize + byteBuffer.position();
		sessionContext.write(byteBuffer);
		// TODO: add by zhuam
		// 查询结果派发

	}

	/**
	 * lazy create ByteBuffer only when needed
	 * 
	 * @return
	 */
	private ByteBuffer allocByteBuffer() {
		if (byteBuffer == null) {
			byteBuffer = session.getSessionContext().allocate();
		}
		return byteBuffer;
	}

	/**
	 * select
	 * 
	 * 元数据返回时触发，将header和元数据内容依次写入缓冲区中
	 */
	@Override
	public void fieldEofResponse(IConnectionHolder connectionHolder, byte[] headerBytes, List<byte[]> fieldBytesList, byte[] eofBytes) {

		headerBytes[3] = ++packetId;
		SessionContext sessionContext = session.getSessionContext();
		byteBuffer = sessionContext.writeToByteBuffer(headerBytes, allocByteBuffer());
		for (int i = 0, size = fieldBytesList.size(); i < size; ++i) {
			byte[] fieldBytes = fieldBytesList.get(i);
			fieldBytes[3] = ++packetId;
			// 保存field信息
			FieldPacket fieldPacket = new FieldPacket();
			fieldPacket.read(fieldBytes);
			fieldPacketList.add(fieldPacket);

			byteBuffer = sessionContext.writeToByteBuffer(fieldBytes, byteBuffer);
		}

		fieldCount = fieldPacketList.size();

		eofBytes[3] = ++packetId;
		byteBuffer = sessionContext.writeToByteBuffer(eofBytes, byteBuffer);

		if (isDefaultNodeShowTable) {

			for (String name : shardingTablesSet) {
				RowDataPacket rowDataPacket = new RowDataPacket(1);
				rowDataPacket.add(StringUtil.getBytes(name.toLowerCase()));
				rowDataPacket.packetId = ++packetId;
				byteBuffer = rowDataPacket.write(byteBuffer, sessionContext, true);
			}

		} else if (isDefaultNodeShowFullTable) {

			for (String name : shardingTablesSet) {
				RowDataPacket rowDataPacket = new RowDataPacket(1);
				rowDataPacket.add(StringUtil.getBytes(name.toLowerCase()));
				rowDataPacket.add(StringUtil.getBytes("BASE TABLE"));
				rowDataPacket.packetId = ++packetId;
				byteBuffer = rowDataPacket.write(byteBuffer, sessionContext, true);
			}
		}
	}

	/**
	 * select
	 * 
	 * 行数据返回时触发，将行数据写入缓冲区中
	 */
	@Override
	public void rowResponse(IConnectionHolder connectionHolder, byte[] rowBytes) {

		if (isDefaultNodeShowTable || isDefaultNodeShowFullTable) {
			RowDataPacket rowDataPacket = new RowDataPacket(1);
			rowDataPacket.read(rowBytes);
			String table = StringUtil.string(rowDataPacket.fieldBytesList.get(0), "");
			if (shardingTablesSet.contains(table.toUpperCase())) {
				return;
			}
		}
		rowBytes[3] = ++packetId;

		if (prepared) {
			RowDataPacket rowDataPacket = new RowDataPacket(fieldCount);
			rowDataPacket.read(rowBytes);
			BinaryRowDataPacket binaryRowDataPacket = new BinaryRowDataPacket();
			binaryRowDataPacket.read(fieldPacketList, rowDataPacket);
			binaryRowDataPacket.packetId = rowDataPacket.packetId;
			// binRowDataPk.write(session.getSource());
			/*
			 * [fix bug] : 这里不能直接将包写到前端连接, 因为在fieldEofResponse()方法结束后buffer还没写出,
			 * 所以这里应该将包数据顺序写入buffer(如果buffer满了就写出),然后再将buffer写出
			 */
			byteBuffer = binaryRowDataPacket.write(byteBuffer, session.getSessionContext(), true);
		} else {
			byteBuffer = session.getSessionContext().writeToByteBuffer(rowBytes, allocByteBuffer());
			// session.getSource().write(row);
		}

	}

	@Override
	public void connectionClose(IConnectionHolder connectionHolder, String reason) {
		ErrorPacket errorPacket = new ErrorPacket();
		errorPacket.packetId = ++packetId;
		errorPacket.errorNo = ErrorCode.ER_ERROR_ON_CLOSE;
		errorPacket.errorMessage = StringUtil.getBytes(reason);
		this.backConnectionErr(errorPacket, connectionHolder);

	}

	public boolean isPrepared() {
		return prepared;
	}

	public void setPrepared(boolean prepared) {
		this.prepared = prepared;
	}

	public void clearResources() {

	}

}